package hit.edu.zjc.Operator;


import hit.edu.zjc.Tool.ShowOutPut;
import hit.edu.zjc.marshalling.ByteSerializerDeserializer;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.IOperatorDescriptorRegistry;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;


/*
 * This Class AFK。
 * */
public class FrameFileScanOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
    private static final long serialVersionUID = 1L;

    private String filePath;

    public FrameFileScanOperatorDescriptor(IOperatorDescriptorRegistry spec, String filepath) {
        super(spec, 1, 1);
        recordDescriptors[0] = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});
        this.filePath = filepath;
    }
    public FrameFileScanOperatorDescriptor(IOperatorDescriptorRegistry spec, File file) {
        super(spec, 1, 1);
        recordDescriptors[0] = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});
        this.filePath = file.getAbsolutePath();
    }

    @Override
    public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, final int partition, int nPartitions) {
        return new AbstractUnaryInputUnaryOutputOperatorNodePushable() {
            private InputStream in;
            private ByteBuffer outputBuffer;
            private FrameTupleAppender outputAppender;
        	  private long filelength=0;
            @Override
            public void open() throws HyracksDataException {
                try {
                	 ShowOutPut.show("FrameFileScanner Start");
                    in = new FileInputStream(filePath);
                    outputBuffer = ctx.allocateFrame();
                    outputAppender = new FrameTupleAppender(ctx.getFrameSize());
                    outputAppender.reset(outputBuffer, true);//true 就是清零了。
                    writer.open();
                } catch (FileNotFoundException e) {
                    throw new HyracksDataException(e);
                }
                
                try {
                    //out.write(buffer.array());
                	byte[] data=new byte[ctx.getFrameSize()];
                	int len=in.read(data);
                	while(len>0)
                		{
                		filelength+=len;
                		//outputAppender.append(bytes, offset, length);
                		outputAppender.reset(outputBuffer, false);
                		if (!outputAppender.append(data,0,len)) 
                			{
                             FrameUtils.flushFrame(outputBuffer, writer);
                             outputAppender.reset(outputBuffer, true);
                             outputAppender.append(data,0,len);
                         	}
                		len=in.read(data);
                		}
                	if (outputAppender.getTupleCount() > 0) {
                         FrameUtils.flushFrame(outputBuffer, writer);
                     		}
                	} 
                catch (IOException e) {
                    throw new HyracksDataException(e);
                	}
                close();
                
                
            }

            @Override
            public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
                	ShowOutPut.show("nextFrame invoke");
                	close();
            }

            @Override
            public void fail() throws HyracksDataException {
            	ShowOutPut.show("FrameFileScanner fail");
            	}

            @Override
            public void close() throws HyracksDataException {
                try {
                    in.close();
                    outputAppender = null;
                    outputBuffer = null;
                    writer.close();
                  ShowOutPut.show("close");
                } catch (IOException e) {
                    throw new HyracksDataException(e);
                }
            }
        };
    }
}